#include "server.h"
#include "net.h"
#include <mutex>

void *user_routine(void *p){
  struct server_ctx *ctx = (struct server_ctx *)p;
  int client_num = ctx->param->client_num, i;

  ctx->client_qp_info = (struct qp_info *)malloc(client_num * sizeof(struct qp_info));
  assert(ctx->client_qp);
  mt_set(ctx->mi, "client-next-id", "0", 1);
  for(i = 0; i < client_num; i++){
    get_qp_info(ctx->mi, "user-client", i, ctx->id, &ctx->client_qp_info[i]);
    connect_qp(ctx->param->ib_port, ctx->client_qp[i], &ctx->client_qp_info[i]);
    p_info("connect QP id %d\n", i);
  }
  p_info("user routine quit\n");
  return NULL;
}
struct recv_ring32 *create_recv_ring32(struct ibv_pd *pd, int num){
  void *mem;
  struct recv_ring32 *p = (struct recv_ring32 *)malloc(sizeof(struct recv_ring32));
  assert(p);
  mem = (struct m32 *)memalign(64, num * 32);
  assert(mem);
  p->mem = (struct m32 *)mem;
  p->num = num;
  p->head = 0;
  p->tail = 0;
  p->mr = ibv_reg_mr(pd, (void *)mem, num << 7, MR_PERMISSION);
  assert(p->mr);
  p->lkey = p->mr->lkey;
  return p;
}
void *server_poll_routine(void *p){
  struct server_ctx *ctx = (struct server_ctx *)p;
  struct ibv_cq *cq;
  int n;
  UNUSED(n);
  UNUSED(cq);
  cq = ctx->server_recv_cq;
  for(;;){
  }
  return NULL;
}
void post_recv(unsigned size, int qp_id, struct ibv_qp *qp, unsigned long buf, unsigned lkey){
  struct ibv_sge li = {buf, size, lkey};
  struct ibv_recv_wr *bad_wr, recv_wr;
  int ret;
  memset(&recv_wr, 0, sizeof(struct ibv_recv_wr));
  recv_wr.wr_id = qp_id;
  recv_wr.sg_list = &li;
  recv_wr.num_sge = 1;
  ret = ibv_post_recv(qp, &recv_wr, &bad_wr);
  assert(!ret);
}
void post_recv_one_ring32(struct recv_ring32 *ring, struct ibv_qp *qp, int qp_id){
  assert((ring->head + 1) % ring->num != ring->tail);
  post_recv(32, qp_id, qp, (unsigned long)&ring->mem[ring->head], ring->lkey);
  ++ring->head;
  if(ring->head >= ring->num)ring->head -= ring->num;
}
void *get_recv_ring_addr(struct recv_ring32 *ring){
  return ring->mem + ring->tail;
}
void put_recv_one_ring32(struct recv_ring32 *ring){
  ++ring->tail;
  if(ring->tail >= ring->num)ring->tail -= ring->num;
}
void post_send(unsigned size, int id, struct ibv_qp *qp, unsigned long addr, unsigned lkey){
  struct ibv_sge li = {addr, size, lkey};
  struct ibv_send_wr *bad_wr, send_wr;
  int res;
  memset(&send_wr, 0, sizeof(send_wr));
  send_wr.wr_id = id;
  send_wr.sg_list = &li;
  send_wr.num_sge = 1;
  send_wr.opcode = IBV_WR_SEND;
  send_wr.send_flags = IBV_SEND_SIGNALED;
  res = ibv_post_send(qp, &send_wr, &bad_wr);
  assert(res == 0);
}
void init_server_struct(struct server_ctx *ctx){
  int i;
  //init metadata hashtable
  {
    struct mempool_meta *mm;
    struct pm_super *super = ctx->super;
    ctx->meta_map = new std::unordered_map<unsigned short, struct mempool_meta *>(MAX_MEMPOOL_NUM << 1);
    ctx->free_meta = new std::deque<struct mempool_meta *>();
    assert(ctx->meta_map);
    for(i = 0; i < MAX_MEMPOOL_NUM; i++){
      mm = super->mempool + i;
      if(mm->id){
        (*(ctx->meta_map))[mm->id] = mm;
      }else{
        ctx->free_meta->push_back(mm);
      }
    }
  }
}

std::deque<std::pair<unsigned long, struct req_struct> >write_queue;
std::mutex write_queue_lock;
std::deque<std::pair<unsigned long, struct res_struct> >send_queue;
std::mutex send_queue_lock;

#define POST_SEND(id, res) do { \
            send_queue_lock.lock(); \
            send_queue.push_back({(unsigned long)id, res}); \
            send_queue_lock.unlock(); \
          }while(0);
#define POST_WRITE(id, res) do { \
            write_queue_lock.lock(); \
            write_queue.push_back({(unsigned long)id, res}); \
            write_queue_lock.unlock(); \
          }while(0);

void do_deal(struct server_ctx *ctx, struct req_struct *req, unsigned long qp_id){
  union res_union ures;
  struct res_struct *res;
  struct pm_super *super;
  decltype(ctx->meta_map) meta_map;
  unsigned short mempool_id;
  struct mempool_meta *mm;

  res = &ures.res;
  super = ctx->super;
  meta_map = ctx->meta_map;
  res->req_id = req->req_id;
  UNUSED(super);
  if(req->type == REQ_alloc_dsm){
    mempool_id = req->alloc_dsm.mp_id;
    if(meta_map->find(mempool_id) != meta_map->end()){
      res->status = RES_exist;
    }else if(ctx->free_meta->size() == 0){
      res->status = RES_nomem;
    }else{
      mm = ctx->free_meta->front();ctx->free_meta->pop_front();
      (*meta_map)[mempool_id] = mm;
      mm->id = mempool_id;
      mm->size = req->alloc_dsm.size;
      mm->root = {(unsigned long)0, (unsigned long)0};
      res->status = RES_success;
    }
    POST_SEND(qp_id, *res);
  }else if(req->type == REQ_write_trans_begin || 
          req->type == REQ_write_trans_end ||
          req->type == REQ_page_get_write_addr){
    POST_WRITE(qp_id, *req);
  }else{
    p_die("unknown requst type\n");
  }
}
struct write_trans {
  unsigned long trans_id;
  unsigned long start, end;
};
struct write_trans_info{
  struct mempool_meta *meta;
  std::map<unsigned long, struct write_trans> *trans; //key is start
  std::unordered_map<unsigned long , unsigned long> *id_start_map; //id -> start
  std::deque<std::pair<unsigned long, struct req_struct> > *trans_queue;
};
bool check_overlap(const struct map<unsigned long, struct write_trans> const *trans, unsigned long start, unsigned long end){
  if(trans->empty())return false;
  auto it = trans->lower_bound(start);
  if(it != trans->end()){
    if(it->second.start <= end)return true;
  }
  if(it != trans->begin()){
    --it;
    if(it->second.end >= start)return true;
    ++it;
  }
  return false;
}
//for info print use
std::unordered_map<unsigned short, struct write_trans_info *> *trans_global;
void *writer_routine(void *p){
  struct server_ctx *ctx;
  struct req_struct req;
  struct res_struct res;
  unsigned long qp_id;
  bool flg;
  unsigned short mp_id;
  unsigned long trans_id = 100;
  std::unordered_map<unsigned short, struct write_trans_info *> trans_map;
  std::deque<std::pair<unsigned long, struct req_struct> > ready_queue;
  struct write_trans_info *info;

  ctx = (struct server_ctx *)p;
  decltype(ctx->meta_map) &meta_map = *(ctx->meta_map);
  trans_map = new std::unordered_map<unsigned short, struct write_trans_info *>(10000);
  trans_global = trans_map;
  for(;;){
    flg = false;
    if(!ready_queue.empty()){
      flg = true;
      qp_id = ready_queue.front().first;
      req = ready_queue.front().second;
      ready_queue.pop_front();
    }else{
      write_queue_lock.lock();
      if(!write_queue.empty()){
        flg = true;
        qp_id = write_queue.front().first;
        req = write_queue.front().second;
        write_queue.pop_front();
      }
      write_queue_lock.unlock();
    }
    if(!flg)continue;
    if(req.type == REQ_write_trans_begin){
      Addr start, end;
      start.val = req.trans_write.start;
      end.val = req.trans_write.end;
      mp_id = begin.region_id;
      if(trans_map.find(mp_id) == trans_map.end()){ //create new trans info
        info = trans_map[mp_id] = (struct write_trans_info *)malloc(sizeof(struct write_trans_info));
        assert(meta_map[mp_id] != meta_map.end());
        info->meta = meta_map[mp_id];
        info->trans = new std::map<unsigned long, struct write_trans>();
        info->id_start_map = new std::unordered_map<unsigned long , unsigned long>();
        info->trans_queue = new std::deque<std::pair<unsigned long, struct req_struct> >();
      }else info = trans_map[mp_id];
      //check the trans region
      if(check_overlap(info->trans, start.offset, end.offset)){ //can't start trans now
        info->trans_queue->push_back(make_pair(qp_id, req));
      }else{//start trans now
        struct write_trans tmp;
        tmp.trans_id = (++trans_id) | ((unsigned long)mp_id << 48);
        assert(!(trans_id >> 48));
        tmp.start = start;
        tmp.end = end;
        info->trans->insert(make_pair(tmp.start, tmp));
        info->id_start_map->insert(make_pair(tmp.trans_id, tmo.start));
        res.req_id = req.req_id;
        res.status = RES_success;
        res.commit_id = tmp.trans_id;
        POST_SEND(qp_id, res);
      }
    }else if(req.type == REQ_write_trans_end){
      unsigned long commit_id, start;
      commit_id = req.commit.commit_id;
      mp_id = commit_id >> 48;
      assert(trans_map->find(mp_id) != trans_map->end());
      info = trans_map[mp_id];
      assert(info->id_start_map->find(commit_id) != info->id_start_map->end());
      start = info->id_start_map->find(commit_id)->second;
      info->id_start_map->erase(commit_id);
      res.req_id = req.req_id;
      res.status = RES_success;
      POST_SEND(qp_id, res);
      if(info->id_start_map->empty()){
        //the commit is over, new commit should begin;
        info->trans->clear();
        if(ready_queue.empty())ready_queue.swap(*(info->trans_queue));
        else{
          while(!info->trans_queue->empty()){
            ready_queue.push_back(trans_queue->front());
            info->trans_queue->pop_front();
          }
        }
      }
    }else if(req.type == REQ_page_get_write_addr){

    }
  }
  return NULL;
}


void *send_routine(void *p){
  union res_union * res;
  struct ibv_mr *res_mr;
  struct ibv_wc send_wc;
  struct ibv_qp **qp;
  std::pair<unsigned long, struct res_struct> ele;
  bool flg;

  struct server_ctx *ctx = (struct server_ctx *)p;
  qp = ctx->client_qp;
  res = (union res_union *)memalign(64, sizeof(union res_union));
  res_mr = ibv_reg_mr(ctx->pd, res, 32, MR_PERMISSION);
  for(;;){
    flg = false;
    send_queue_lock.lock();
    if(!send_queue.empty()){
      ele = send_queue.front();
      send_queue.pop_front();
      flg = true;
    }
    send_queue_lock.unlock();
    if(!flg)continue;
    *res = *(union res_union *)(&ele.second);
    post_send(32, ele.first, qp[ele.first], (unsigned long)res, res_mr->lkey);
    while(!ibv_poll_cq(ctx->client_send_cq[ele.first], 1, &send_wc));
    assert(send_wc.status == 0);
  }
  return NULL;
}
void *client_poll_routine(void *p){
  struct server_ctx *ctx = (struct server_ctx *)p;
  struct ibv_cq *cq;
  struct ibv_qp **qp;
  int client_num;
  int n, i ,j;
  struct recv_ring32 **rr;
  struct ibv_wc wc[POLL_NUM];
  union req_union req;
  pthread_t write_thread, send_thread;
  
  client_num = ctx->param->client_num;
  cq = ctx->client_recv_cq;
  qp = ctx->client_qp;
  pthread_create(&write_thread, NULL, writer_routine, ctx);
  pthread_create(&send_thread, NULL, send_routine, ctx);
  rr = (struct recv_ring32 **)malloc(client_num * sizeof(struct recv_ring32 *));
  for(i = 0; i < client_num; i++)rr[i] = create_recv_ring32(ctx->pd, 64);
  for(i = 0; i < client_num; i++){
    for(j = 0;j < 16; j++){ //16 post recv ahead
      post_recv_one_ring32(rr[i], qp[i], i);
    }
  }
  for(;;){
    n = ibv_poll_cq(cq, POLL_NUM, wc);
    for(i = 0; i < n; i++){
      if(wc[i].status){
        p_die("status wc %d\n", wc[i].status);
      }
      //p_info("get wc, id: %d type %d \n\n", (int)wc[i].wr_id, (int)wc[i].opcode);
      req = *(union req_union *) get_recv_ring_addr(rr[wc[i].wr_id]);
      put_recv_one_ring32(rr[wc[i].wr_id]);
      do_deal(ctx, (struct req_struct *)&req, wc[i].wr_id);
    }
  }
  
  return NULL;
}
 /*
 struct write_trans {
  unsigned long trans_id;
  unsigned long start, end;
};
struct write_trans_info{
  struct mempool_meta *meta;
  std::map<unsigned long, struct write_trans> *trans; //key is start
  std::unordered_map<unsigned long , unsigned long> *id_start_map; //id -> start
  std::deque<std::pair<unsigned long, struct req_struct> > *trans_queue;
};
 */
void main_control(struct server_ctx *ctx){
  char input[256];
  for(;;){
    p_info(">>");
    scanf("%s", input);
    if(strcmp(input, "ls") == 0){
      p_info("total %ld file\n", ctx->meta_map->size());;
      for(auto &x : *(ctx->meta_map)){
        p_info("id %u size %ld\n", x.first, x.second->size);
      }
    }else if(strcmp(input, "tx") == 0){
      for(auto &info : *trans_global){
        p_info("mempool info %u\n", info.first);
        p_info("all tx exists: \n")
        for(auto &t : *(info.second->trans)){
          p_info("  tx id %lx  start %ld end %ld    %s\n", t.second->trans_id, 
                      t.second->start, t.second->end, 
                      info.id_start_map->find(t.second->trans_id) == info.id_start_map->end() ? "DONE" : "LIVE");
        }
        p_info("wait trans: \n");
        for(auto &w : *info.trans_queue){
          p_info("qp_id %d ")
        }
      }
    }else{
      p_info("unknown commond\n");
    }
  }
}